package com.ionehe.group.common.notify.rocketmq.producer;

import com.ionehe.group.common.notify.message.producer.TransactionProducer;
import com.ionehe.group.common.notify.rocketmq.flags.RocketKeyFlags;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【rocketmq生产者(事务型消息)】
 */
public class TransactionProducerImpl implements TransactionProducer {
    TransactionMQProducer transactionMQProducer;
    private Properties properties;
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean closed = new AtomicBoolean(false);

    public TransactionProducerImpl(Properties properties, TransactionCheckListener transactionCheckListener) {
        this.properties = properties;
        this.transactionMQProducer = new TransactionMQProducer((String) RocketKeyFlags.FLAG_PRODUCER_GROUP.get(properties));
        this.transactionMQProducer.setNamesrvAddr((String) RocketKeyFlags.FLAG_NAMESRVADDR.get(properties));
        this.transactionMQProducer.setTransactionCheckListener(transactionCheckListener);
    }


    @Override
    public void start() {
        if (this.started.compareAndSet(false, true)) {
            if (this.transactionMQProducer.getTransactionCheckListener() == null) {
                throw new IllegalArgumentException("TransactionCheckListener is null");
            }
            try {
                this.transactionMQProducer.start();
            } catch (MQClientException e) {
                throw new RuntimeException(e);
            }
        }
    }


    @Override
    public void shutdown() {
        if (this.closed.compareAndSet(false, true)) {
            this.transactionMQProducer.shutdown();
        }
    }

    @Override
    public SendResult send(Message message, LocalTransactionExecuter executer, Object arg) {
        TransactionSendResult transactionSendResult;
        MessageAccessor.putProperty(message, RocketKeyFlags.FLAG_PRODUCER_GROUP.getKey(), (String) RocketKeyFlags.FLAG_CONSUMER_GROUP.get(this.properties));

        try {
            transactionSendResult = this.transactionMQProducer.sendMessageInTransaction(message, executer, arg);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return (SendResult) transactionSendResult;
    }


    @Override
    public boolean isStarted() {
        return this.started.get();
    }


    @Override
    public boolean isClosed() {
        return this.closed.get();
    }
}


